package com.bw.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.base.BaseApp;
import com.bw.gmall.realtime.common.bean.TradePaymentBean;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.function.DorisMapFunction;
import com.bw.gmall.realtime.common.util.DateFormatUtil;
import com.bw.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Iterator;

public class DwsTradePaymentSucWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTradePaymentSucWindow().start(Constant.TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS,Constant.DWS_TRADE_PAYMENT_SUC_WINDOW,4,10027);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {

        //etl
//        dataStreamSource.print();

        SingleOutputStreamOperator<JSONObject> streamOperator = eltStream(dataStreamSource);

        //水位线
        SingleOutputStreamOperator<JSONObject> waterStream = WaterStream(streamOperator);

        //聚合
        SingleOutputStreamOperator<TradePaymentBean> streamOperator1 = UV_newpayStream(waterStream);

//        streamOperator1.print("1");


        //开窗统计
        SingleOutputStreamOperator<TradePaymentBean> windowStream = WindowStream(streamOperator1);

        windowStream.print();
        //写入doris
        windowStream.map(new DorisMapFunction<>()).sinkTo(FlinkSinkUtil.getDorisSinks(Constant.DWS_TRADE_PAYMENT_SUC_WINDOW));

    }

    private static SingleOutputStreamOperator<TradePaymentBean> WindowStream(SingleOutputStreamOperator<TradePaymentBean> streamOperator1) {
         return streamOperator1.windowAll(TumblingEventTimeWindows.of(Time.seconds(1)))
                .reduce(new ReduceFunction<TradePaymentBean>() {
                    @Override
                    public TradePaymentBean reduce(TradePaymentBean t1, TradePaymentBean t2) throws Exception {
                        t1.setPaymentSucNewUserCount(t1.getPaymentSucNewUserCount() + t2.getPaymentSucNewUserCount());
                        t1.setPaymentSucUniqueUserCount(t1.getPaymentSucUniqueUserCount() + t2.getPaymentSucUniqueUserCount());

                        return t1;
                    }
                }, new AllWindowFunction<TradePaymentBean, TradePaymentBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow timeWindow, Iterable<TradePaymentBean> iterable, Collector<TradePaymentBean> collector) throws Exception {

                        long start = timeWindow.getStart();

                        long end = timeWindow.getEnd();

                        Iterator<TradePaymentBean> iterator = iterable.iterator();

                        TradePaymentBean tradePaymentBean = iterator.next();

                        tradePaymentBean.setStt(DateFormatUtil.tsToDateTime(start));
                        tradePaymentBean.setEdt(DateFormatUtil.tsToDateTime(end));
                        tradePaymentBean.setCurDate(DateFormatUtil.tsToDateTime(System.currentTimeMillis()));

                        collector.collect(tradePaymentBean);
                    }
                });
    }

    private static SingleOutputStreamOperator<TradePaymentBean> UV_newpayStream(SingleOutputStreamOperator<JSONObject> waterStream) {
         return waterStream.keyBy(x -> x.getString("user_id")).process(new ProcessFunction<JSONObject, TradePaymentBean>() {

            ValueState<String> state;


            @Override
            public void open(Configuration parameters) {
                ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("state", String.class);
                state = getRuntimeContext().getState(stateDescriptor);

            }

            @Override
            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, TradePaymentBean>.Context context, Collector<TradePaymentBean> collector) throws Exception {

                Long ts = jsonObject.getLong("ts");

                String tsToDate = DateFormatUtil.tsToDateTime(ts);

                String value = state.value();

                long uv = 0L;
                long newcv = 0L;

                //若 状态和 时间不相等，表示是今日第一次 支付
                if (!tsToDate.equals(value)) {
                    uv = 1L;
                    state.update(tsToDate);

                    // 若 value==null 代表 从未支付过 为新用户
                    if (value == null) {
                        newcv = 1L;
                    }
                }

                if (uv == 1) {

                    collector.collect(TradePaymentBean.builder()
                            .paymentSucUniqueUserCount(uv)
                            .paymentSucNewUserCount(newcv)
                            .curDate(tsToDate)
                                    .ts(ts)
                            .build());
                }
            }
        });
    }

    private static SingleOutputStreamOperator<JSONObject> WaterStream(SingleOutputStreamOperator<JSONObject> streamOperator) {
         return streamOperator.assignTimestampsAndWatermarks(WatermarkStrategy
                .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                .withTimestampAssigner((event, timestamp) -> event.getLong("ts")).withIdleness(Duration.ofSeconds(1)));
    }

    private static SingleOutputStreamOperator<JSONObject> eltStream(DataStreamSource<String> dataStreamSource) {
         return dataStreamSource.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String s, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {

                if (s != null) {

                    JSONObject jsonObject = JSON.parseObject(s);

                    String user_id = jsonObject.getString("user_id");

                    //ts为10位
                    Long ts = jsonObject.getLong("ts");


                    if (user_id != null && ts != null) {

                        jsonObject.put("ts", ts * 1000);

                        collector.collect(jsonObject);
                    }

                }
            }
        });
    }
}
